package main

import (
	"context"
	"fmt"
	"os"

	"github.com/louloulin/dataflare/pkg/workflow/connector"
)

func main() {
	// Create a connector registry
	registry := connector.NewConnectorRegistry()
	registry.RegisterBuiltinConnectors()

	// Create a connector manager
	manager := connector.NewConnectorManager(registry.GetFactory())

	// Create a memory source connector
	err := manager.CreateSourceConnector("memory-source", "memory", map[string]interface{}{
		"data": []interface{}{
			map[string]interface{}{"id": "1", "value": 10},
			map[string]interface{}{"id": "2", "value": 20},
			map[string]interface{}{"id": "3", "value": 30},
		},
	})
	if err != nil {
		fmt.Printf("Failed to create memory source connector: %v\n", err)
		os.Exit(1)
	}

	// Create a console sink connector
	err = manager.CreateSinkConnector("console-sink", "console", map[string]interface{}{
		"format": "json",
	})
	if err != nil {
		fmt.Printf("Failed to create console sink connector: %v\n", err)
		os.Exit(1)
	}

	// Create a file sink connector
	err = manager.CreateSinkConnector("file-sink", "file", map[string]interface{}{
		"path":   "output.json",
		"format": "json",
	})
	if err != nil {
		fmt.Printf("Failed to create file sink connector: %v\n", err)
		os.Exit(1)
	}

	// Start the connectors
	err = manager.StartSourceConnector("memory-source")
	if err != nil {
		fmt.Printf("Failed to start memory source connector: %v\n", err)
		os.Exit(1)
	}

	err = manager.StartSinkConnector("console-sink")
	if err != nil {
		fmt.Printf("Failed to start console sink connector: %v\n", err)
		os.Exit(1)
	}

	err = manager.StartSinkConnector("file-sink")
	if err != nil {
		fmt.Printf("Failed to start file sink connector: %v\n", err)
		os.Exit(1)
	}

	// Get the source connector
	source, err := manager.GetSourceConnector("memory-source")
	if err != nil {
		fmt.Printf("Failed to get memory source connector: %v\n", err)
		os.Exit(1)
	}

	// Get the sink connectors
	consoleSink, err := manager.GetSinkConnector("console-sink")
	if err != nil {
		fmt.Printf("Failed to get console sink connector: %v\n", err)
		os.Exit(1)
	}

	fileSink, err := manager.GetSinkConnector("file-sink")
	if err != nil {
		fmt.Printf("Failed to get file sink connector: %v\n", err)
		os.Exit(1)
	}

	// Read from the source
	ctx := context.Background()
	batch, err := source.Read(ctx)
	if err != nil {
		fmt.Printf("Failed to read from memory source connector: %v\n", err)
		os.Exit(1)
	}

	// Process the records
	for i, record := range batch.Records {
		fmt.Printf("Processing record %d: %s\n", i+1, string(record.Value))
	}

	// Write to the sinks
	fmt.Println("\nWriting to console sink:")
	err = consoleSink.Write(ctx, batch)
	if err != nil {
		fmt.Printf("Failed to write to console sink connector: %v\n", err)
		os.Exit(1)
	}

	fmt.Println("\nWriting to file sink:")
	err = fileSink.Write(ctx, batch)
	if err != nil {
		fmt.Printf("Failed to write to file sink connector: %v\n", err)
		os.Exit(1)
	}

	// Flush the sinks
	err = consoleSink.Flush(ctx)
	if err != nil {
		fmt.Printf("Failed to flush console sink connector: %v\n", err)
		os.Exit(1)
	}

	err = fileSink.Flush(ctx)
	if err != nil {
		fmt.Printf("Failed to flush file sink connector: %v\n", err)
		os.Exit(1)
	}

	// Stop the connectors
	err = manager.StopSourceConnector("memory-source")
	if err != nil {
		fmt.Printf("Failed to stop memory source connector: %v\n", err)
		os.Exit(1)
	}

	err = manager.StopSinkConnector("console-sink")
	if err != nil {
		fmt.Printf("Failed to stop console sink connector: %v\n", err)
		os.Exit(1)
	}

	err = manager.StopSinkConnector("file-sink")
	if err != nil {
		fmt.Printf("Failed to stop file sink connector: %v\n", err)
		os.Exit(1)
	}

	fmt.Println("\nConnector example completed successfully!")
	fmt.Println("Output written to output.json")
}
